package KafkaApps;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.LongDeserializer;

import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Collections;
import java.util.Properties;

public class ConsumerFileWhole {
    private final static String TOPIC = "testTrans";
    private final static String FILEPATH = "trans_file/consumer_file/afterkafka.jpg";

    public static Consumer<Long, byte[]> createConsumer() {
        final Properties sets = new Properties();
        sets.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "master:9092");
        sets.put(ConsumerConfig.GROUP_ID_CONFIG, "transFile");
        sets.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, ConsumerConfig.DEFAULT_FETCH_MAX_BYTES);
        sets.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,LongDeserializer.class.getName());
        sets.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());

        final Consumer<Long, byte[]> consumer = new KafkaConsumer<>(sets);
        consumer.subscribe((Collections.singletonList(TOPIC)));
        return consumer;
    }

    static void runConsumer(){
        final Consumer<Long, byte[]> consumer = createConsumer();

        final int giveUp = 100;   int noRecordsCount = 0;
        FileOutputStream fileOutputStream = null;

        while (true) {
            final ConsumerRecords<Long, byte[]> consumerRecords =
                    consumer.poll(1000);

            if (consumerRecords.count()==0) {
                noRecordsCount++;
                if (noRecordsCount > giveUp) break;
                else continue;
            }

            for (ConsumerRecord<Long, byte[]> consumerRecord: consumerRecords) {
                String fileName = "trans_file/consumer_file/after_" + consumerRecord.offset() + ".jpg";
                System.out.printf("Consumer Record:%s(%d, , %d, %d)\n",fileName,
                        consumerRecord.key(), consumerRecord.partition(), consumerRecord.offset());
                try {
                    fileOutputStream = new FileOutputStream(fileName);
                    fileOutputStream.write(consumerRecord.value());
                } catch (FileNotFoundException e) {
                    e.printStackTrace();
                } catch (IOException e) {
                    e.printStackTrace();
                } finally {
                    try {
                        fileOutputStream.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }


            consumer.commitAsync();
        }
        consumer.close();
        System.out.println("DONE");
    }


    public static void main(String... args) throws Exception {
        runConsumer();
    }
}
